package com.peng.sparktest.sparkstreaming

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object StreamTest02_DStreamApiTest01 {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("dstreamApiTest")

    val context = new SparkContext(conf)

    context.setLogLevel("ERROR")


    /**
     *
     * 1、window
     *
     * 首先从一个需求引出知识点
     * 假设有一个需求：
     * 需要对一批数据，既按1秒的频率进行统计，又要按5秒的频率进行统计，应该怎么做？
     *
     */


    //此处我们以1秒的频率获取批数据
    //获取的粒度一定是尽量小的，之后的操作只能操作比此粒度更粗的数据
    val sc = new StreamingContext(context, Duration(1000)) //Duration和Second的区别是，一个是毫秒，一个是秒

    val dStream: ReceiverInputDStream[String] = sc.socketTextStream("localhost", 8888)

    //首先按1秒频率进行输出
    //    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), item(1).toInt))
    //    mapData.mapPartitions(iter=>{println("1s");iter}).print()

    //要想实现上面的需求，需要用到window 窗口的概念

    //   val res: DStream[(String, Int)] = mapData.window(Duration(5000)) //存取至今为止5秒内的历史数据，窗口步长不设置就按context设置的拉取频率进行滑动,但似乎并不能实现我们的需求
    //    res.print()

    //原来我们还少设置了一个参数，窗口的步长！   这样就能实现按5秒内数据进行输出
    //    val res: DStream[(String, Int)] = mapData.window(
    //      Duration(5000), //窗口的记录历史近几秒的数据批，注意该值需要为获取批数据时间的整数倍
    //      Duration(5000) //窗口的滑动（步长）大小
    //    )
    //    res.print()


    /**
     * 接下来进一步体会window两参数不同的各种用法
     * 进行对几秒内数据进行统计word count
     */
    //    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), 1))

    //每5秒收集一次历史1秒内的数据批
    //    val res: DStream[(String, Int)] = mapData.window(
    //      Duration(1000), //窗口的记录历史近几秒的数据批
    //      Duration(5000) //窗口的滑动（步长）大小
    //    )

    //每1秒收集一次历史5秒内的数据批
    //    val res: DStream[(String, Int)] = mapData.window(
    //      Duration(5000), //窗口的记录历史近几秒的数据批
    //      Duration(1000) //窗口的滑动（步长）大小
    //    )

    //每10秒收集一次历史5秒内的数据批
    //    val res: DStream[(String, Int)] = mapData.window(
    //      Duration(5000), //窗口的记录历史近几秒的数据批
    //      Duration(10000) //窗口的滑动（步长）大小
    //    )
    //
    //    val reduceRes: DStream[(String, Int)] = res.reduceByKey(_+_)
    //    reduceRes.print()

    //上面的都是先window，再统计，那如果翻转过来呢？
    //会发现，语义发生变化了，这种写法意味着：
    //将历史5秒内（每秒的reduce后的转换结果集）作为批数据 进行累积
    //    val reduceRes: DStream[(String, Int)] = mapData.reduceByKey(_ + _)
    //    //这里只是负责了收集每一秒的批数据，而并未对5秒的数据批做统计操作
    //    val res: DStream[(String, Int)] = reduceRes.window(
    //      Duration(5000), //窗口的记录历史近几秒的数据批
    //      Duration(1000) //窗口的滑动（步长）大小
    //    )

    //为了防范这种可能有不符合预期的调用，其实有一个可以代替上面操作的方法：
    //    val res: DStream[(String, Int)] = mapData.reduceByKeyAndWindow(
    //      (old: Int, n: Int) => old + n, //reduce func
    //      Duration(5000), //窗口的记录历史近几秒的数据批
    //      Duration(1000) //窗口的滑动（步长）大小
    //    )
    //
    //    res.print()


    /**
     * 2、内部RDD编程调用机会
     *
     * 有时候Dstream自身的带有一些算子可能不够丰富，不能满足我们的需要，我们更需要像之前的RDD那样，
     * low level api  自定制性更高
     */

    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), 1))

    //转成RDD的方式有两种：

    //1、调用transform，且最后必须要返回一个RDD  重新包入到DStream里 相当于一个转换算子
    //    val dRes: DStream[(String, Int)] = mapData.transform((rdd: RDD[(String, Int)]) => {
    //      rdd.foreach(println) //触发job
    //      //像这样，就可以以RDD方式对我们的数据进行中间处理并返回
    //      val resRDD: RDD[(String, Int)] = rdd.map((x: (String, Int)) =>(x._1,x._2*10))
    //      resRDD
    //    })
    //    dRes.print() //触发job

    //2、调用foreach 无需返回任何东西  末端处理，相当于我们定制自己的action执行逻辑
    mapData.foreachRDD((rdd: RDD[(String, Int)]) => {
      rdd.foreach(x => println(s"hahaha~${x}"))
    })
    mapData.print()


    /**
     * 3、作用域：Application，Job，Record
     */
    //    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), 1))
    //
    //    println("Application~") //Application级别 只会打印一次 是在driver端执行
    //
    //    mapData.foreachRDD(rdd=>{
    //      println("Job~") //job级别，每个job执行一次 是在driver端执行
    //      rdd.foreach(record=>{
    //        println("Record~") //一条记录打印一次  在executor端执行
    //
    //      })
    //    })

    /**
     * 4、不同情况的广播变量
     */
    //    val mapData: DStream[(String, Int)] = dStream.map(_.split(" ")).map(item => (item(0), item(1).toInt))
    //
    //    var jobNum = 0
    //    var matchIds: Broadcast[List[Int]] = null
    //    val transDStream: DStream[(String, Int)] = mapData.transform(rdd => {
    //      jobNum += 1
    //      if (jobNum <= 5) {
    //        matchIds = sc.sparkContext.broadcast((1 to 10).toList)
    //      } else {
    //        matchIds = sc.sparkContext.broadcast((11 to 15).toList)
    //
    //      }
    //
    //      rdd.filter(x => matchIds.value.contains(x._2))
    //    })
    //
    //    transDStream.print()


    sc.start()
    sc.awaitTermination()


  }

}
